package com.intct.utils;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

/**
 * @author gufg
 * @since 2025-10-25 22:34
 */
public class FlinkChenageLogConverUtil {
    public static SingleOutputStreamOperator<Row> changeLogConver(DataStream<Row> rowDataStream, String fieldName) {
        return rowDataStream.filter(row -> {
            // 获取数据中的操作类型
            RowKind rowKind = row.getKind();
            // 获取真实数的操作类型
            String rowkKindStr = rowKind.shortString();

            // 将真实数据中的数据变更改为增加
            row.setKind(RowKind.INSERT);

            // 将对应用操作类型转换到特殊字段
            if ("+I".equals(rowkKindStr)) {
                row.setField(fieldName, RowKind.INSERT.shortString());
            } else if ("-U".equals(rowkKindStr)) {
                row.setField(fieldName, RowKind.UPDATE_BEFORE.shortString());
            } else if ("+U".equals(rowkKindStr)) {
                row.setField(fieldName, RowKind.UPDATE_AFTER.shortString());
            } else {
                row.setField(fieldName, RowKind.DELETE.shortString());
            }
            return true;
        });
    }
}
